package cn.monkey.transport.core;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public abstract class AbstractReactiveExchangeManager<C, EXCHANGE extends Exchange> implements ReactiveExchangeManager<C, EXCHANGE> {
    protected final ReactiveExchangeFactory<C, EXCHANGE> factory;
    protected final Map<String, EXCHANGE> exchangeMap;

    public AbstractReactiveExchangeManager(ReactiveExchangeFactory<C, EXCHANGE> factory,
                                           Map<String, EXCHANGE> exchangeMap) {
        this.factory = factory;
        this.exchangeMap = exchangeMap;
    }

    public AbstractReactiveExchangeManager(ReactiveExchangeFactory<C, EXCHANGE> factory) {
        this(factory, new ConcurrentHashMap<>());
    }

    @Override
    public Mono<EXCHANGE> findOrCreate(String key, C c) {
        return Mono.just(this.exchangeMap.get(key))
                .switchIfEmpty(Mono.defer(() -> this.factory.apply(key, null, c)))
                .map(exchange -> this.exchangeMap.compute(key, (k, v) -> {
                    if (v == null || !v.isActive()) {
                        return exchange;
                    }
                    return v;
                }));
    }

    @Override
    public Mono<Void> destroy(EXCHANGE exchange) {
        Iterator<Map.Entry<String, EXCHANGE>> iterator = this.exchangeMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, EXCHANGE> e = iterator.next();
            Exchange value = e.getValue();
            if (value == exchange) {
                iterator.remove();
                return value.destroy();
            }
            if (value instanceof DelegatesExchange delegatesExchange) {
                Collection<Exchange> delegates = delegatesExchange.delegates();
                if (!delegates.contains(exchange)) {
                    continue;
                }
                boolean b = delegates.stream().noneMatch(Exchange::isActive);
                if (b) {
                    iterator.remove();
                }
                return Flux.merge(delegatesExchange.destroy(exchange)).then();
            }
        }
        return Mono.error(new IllegalArgumentException("exchange not exists"));
    }
}
